package com.haizhi.thread;


import com.haizhi.config.BeanstalkConf;
import com.haizhi.task.SchedulerWorker;
import com.haizhi.util.ConfUtil;
import com.haizhi.util.PropertyUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: youfeng
 * Date: 2018-06-25
 * Time: 18:06
 */
public class DispatchThread extends TaskThread {
    private static final Logger logger = LogManager.getLogger(DispatchThread.class);

    //允许调度省份控制
    private Set<String> schedulerProvinceSet;

    //消息队列阈值
    private int beanstalkThresholdNum;

    //调度器
    private Map<String, SchedulerWorker> schedulerWorkerMap = new HashMap<>();

    public DispatchThread() throws Exception {
        super();

        // 提取可以调度的省份信息
        schedulerProvinceSet = ConfUtil.getConfSet("scheduler_province.properties");
        schedulerProvinceSet.forEach(province -> logger.info("当前被调度省份: {}", province));

        //消息队列阻塞阈值
        beanstalkThresholdNum = PropertyUtil.getInt("beanstalk.threshold.num");

        //初始化消息队列配置信息
        InitSchedulerWorker("beanstalk/producer.properties");
    }

    //初始化消息队列
    private void InitSchedulerWorker(String configPath) throws Exception {
        Properties properties = ConfUtil.getProperties(configPath);
        if (null == properties) {
            throw new Exception("初始化生产者配置异常");
        }
        schedulerProvinceSet.forEach(province -> {

            String host = properties.getProperty("beanstalk.host." + province);
            int port = Integer.valueOf(properties.getProperty("beanstalk.port." + province));
            String tube = properties.getProperty("beanstalk.tube." + province);

            BeanstalkConf beanstalkConf = new BeanstalkConf(host, port, tube);
            SchedulerWorker schedulerWorker = new SchedulerWorker(province,
                    beanstalkThresholdNum, beanstalkConf, mongoDatabase, redisHandler);
            schedulerWorkerMap.put(province, schedulerWorker);
            logger.info("初始化调度器完成: province = {}", province);
        });
        logger.info("所有调度器初始化完成...");
    }

    @Override
    public void run() {
        logger.info("启动分发请求线程...");

        while (isRunning()) {
            AtomicBoolean isScheduleSuccess = new AtomicBoolean(false);
            schedulerWorkerMap.forEach((province, worker) -> {
                if (worker.schedule()) {
                    isScheduleSuccess.set(true);
                }
            });
            if (!isScheduleSuccess.get()) {
                logger.info("所有省份都未调度, 休眠10s");
                try {
                    Thread.sleep(10 * 1000);
                } catch (InterruptedException e) {
                    logger.error("休眠被打断: ", e);
                }
            }
        }

        close();
        logger.info("退出分发请求线程...");
    }

    //关闭资源
    private void close() {
        schedulerWorkerMap.forEach((s, schedulerWorker) -> {
            schedulerWorker.close();
        });
        logger.info("所有资源释放完成...");
    }
}
